
/**
* 上海中赢金融信息服务有限公司
* Copyright (c) 2017, chinazyjr All Rights Reserved.
*/
package com.sys.kafka.core.consumer.listener;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import com.sys.kafka.factory.ConsumerFactory;
import com.sys.midware.logger.Logger;
import com.sys.midware.logger.LoggerFactory;

/**
 * <b>Description：</b> TODO <br/>
 * <b>ClassName：</b> ConsumerHandler <br/>
 * <b>@author：</b> mobing <br/>
 * <b>@date：</b> 2017年7月13日 下午4:34:36 <br/>
 * <b>@version: </b> <br/>
 */
public class ConsumerExecute {
    private static final Logger logger = LoggerFactory.getLogger(ConsumerFactory.class);
    private  KafkaConsumer<String, String> consumer;

    private ExecutorService executors;

    public ConsumerExecute(final KafkaConsumer<String, String> consumer){
        this.consumer = consumer;
    }
    
    
    public void execute(int workerNum) {
//        executors = new ThreadPoolExecutor(workerNum, workerNum, 0L,TimeUnit.MILLISECONDS,
//                new ArrayBlockingQueue(1000), new ThreadPoolExecutor.CallerRunsPolicy());
        
        executors = Executors.newCachedThreadPool();
        
        
       final KafkaConsumer<String, String> consumer =  this.consumer ;
        Thread t = new Thread(new Runnable() {// 启动一个子线程来监听kafka消息
            public void run() {
                while (true) {
                    ConsumerRecords<String, String> records = consumer.poll(200);
                    for (final ConsumerRecord<String, String> record : records) {
                        System.out.println("监听到kafka消息。。。。。。record="  + record);
                        executors.submit(new ConsumerWorker(record));
                    }
                }
            }
        });
        t.start();

    }
    
    public void shutdown() {
        if (consumer != null) {
            consumer.close();
        }
        if (executors != null) {
            executors.shutdown();
        }
        try {
            if (!executors.awaitTermination(10, TimeUnit.SECONDS)) {
                System.out.println("Timeout.... Ignore for this case");
            }
        } catch (InterruptedException ignored) {
            System.out.println("Other thread interrupted this shutdown, ignore for this case.");
            Thread.currentThread().interrupt();
        }
    }
}
